package com.hexinfo.dmpro.component.comparison.handler.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.hexinfo.dmpro.component.common.entity.TableFieldsEntity;
import com.hexinfo.dmpro.component.common.exception.BizException;
import com.hexinfo.dmpro.component.common.service.impl.ImpalaQueryService;
import com.hexinfo.dmpro.component.common.util.DataSourceUtil;
import com.hexinfo.dmpro.component.comparison.constants.DataContrastConstants;
import com.hexinfo.dmpro.component.comparison.entity.TableContrastDataManagementEntity;
import com.hexinfo.dmpro.component.comparison.entity.TableContrastParamEntity;
import com.hexinfo.dmpro.component.comparison.handler.IDataReadHandler;
import com.hexinfo.dmpro.component.comparison.handler.ITableContrastHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;

@Slf4j
@Service
public class TableContrastHandler implements ITableContrastHandler {

    @Resource
    private TableContrastDataManagementEntity dataManagement;

    @Resource
    private IDataReadHandler dataReadHandler;

    @Resource
    private ImpalaQueryService impalaQueryService;

    @Resource
    private DataContrastConstants dataContrastConstants;


    /**
     * @Method createTableContrastTask
     * @Param tableContrastParam
     * @Return java.lang.Boolean
     * @Description 表对比
     * @Author Huang Yanyihe
     * @Date 2023/9/6 16:33
     * @Version V1.0
     */
    @Override
    @Async("tableContrastAsyncTaskExecutor")
    public void startTableDataContrast(TableContrastParamEntity tableContrastParam) {
        log.info("[数据比对组件]- 表[{}]数据比对任务开始, 开始时间:{}", tableContrastParam.getTableFlag(), DateUtil.now());
        try {
            // ① 检查源表和目标表是否存在
            checkTableExist(tableContrastParam);
            // ② 检查源表和目标表数据总量是否一致(若数据量为0则不需要继续对比)
            long dataTotal = checkDataCount(tableContrastParam);
//            if (dataTotal > 0) {
//                // ③ 检查统计数据是否一致
//                checkStatisticsData(tableContrastParam);
//                // ④ 检查表数据明细是否一致
//                checkDetailedData(tableContrastParam, dataTotal);
//            } else {
            dataManagement.setAchieveStatus(tableContrastParam.getTableFlag());
//            }
        } catch (BizException e) {
            log.warn("[数据对比组件] 表[{}]数据比对异常:{}", tableContrastParam.getTableFlag(), e.getErrMsg());
            tableContrastFailHandler(tableContrastParam, e.getErrMsg());
        } catch (Exception e) {
            log.error("[数据对比组件] 表[{}]数据比对异常:{}", tableContrastParam.getTableFlag(), e.getMessage(), e);
            tableContrastFailHandler(tableContrastParam, e.getMessage());
        } finally {
            log.info("[数据比对组件]- 表[{}]数据比对任务结束, 结束时间:{}", tableContrastParam.getTableFlag(), DateUtil.now());
        }
    }

    /**
     * @Method checkTableExist
     * @Param tableContrastParam
     * @Return boolean
     * @Description 检查表存在
     * @Author Huang Yanyihe
     * @Date 2023/9/10 22:01
     * @Version V1.0
     */
    private void checkTableExist(TableContrastParamEntity tableContrastParam) {
        log.info("[数据比对组件]- 表[{}]是否存在对比开始,开始时间: {}", tableContrastParam.getTableFlag(), DateUtil.now());
        isSourceTableExist(tableContrastParam);
        isTargetTableExist(tableContrastParam);
        log.info("[数据比对组件]- 表[{}]是否存在对比结束,结束时间: {}", tableContrastParam.getTableFlag(), DateUtil.now());
    }

    /**
     * @Method isTargetTableExist
     * @Param tableContrastParam
     * @Return void
     * @Description 判断源表是否存在
     * @Author Huang Yanyihe
     * @Date 2023/9/18 13:29
     * @Version V1.0
     */
    private void isSourceTableExist(TableContrastParamEntity paramEntity) {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getImpalaSourceConnection();
            impalaQueryService.isTableExist(connection, paramEntity.getSourceDbName(), paramEntity.getSourceTableName());
        } catch (BizException e) {
            throw new BizException(e.getErrMsg());
        } catch (Exception e) {
            throw new RuntimeException(StrUtil.format("源表是否存在检查异常: {}", e.getMessage()));
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    /**
     * @Method isTargetTableExist
     * @Param tableContrastParam
     * @Return void
     * @Description 判断目标表是否存在
     * @Author Huang Yanyihe
     * @Date 2023/9/18 13:29
     * @Version V1.0
     */
    private void isTargetTableExist(TableContrastParamEntity paramEntity) {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getImpalaTargetConnection();
            impalaQueryService.isTableExist(connection, paramEntity.getSourceDbName(), paramEntity.getSourceTableName());
        } catch (BizException e) {
            throw new BizException(e.getErrMsg());
        } catch (Exception e) {
            throw new RuntimeException(StrUtil.format("对比表是否存在检查异常: {}", e.getMessage()));
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    /**
     * @Method checkDataCount
     * @Param tableContrastParam
     * @Return boolean
     * @Description 数据总量比对
     * @Author Huang Yanyihe
     * @Date 2023/9/10 22:12
     * @Version V1.0
     */
    private long checkDataCount(TableContrastParamEntity params) {
        log.info("[数据比对组件]- 表[{}]表数据总量对比开始,开始时间: {}", params.getTableFlag(), DateUtil.now());
        // ① 获取源表数据总量
        long sourceDataNum = getSourceTableCount(params);
        // ② 获取比对表数据总量
        long targetDataNum = getTargetTableCount(params);
        // ③ 判断两表数据总量是否一致
        dataTotalCompare(params, sourceDataNum, targetDataNum);
        log.info("[数据比对组件]- 表[{}]表数据总量对比结束,结束时间: {}, 数据总量: {}", params.getTableFlag(), DateUtil.now(), sourceDataNum);
        return sourceDataNum;
    }

    /*
     * @Method: getSourceTableCount <br>
     * @Param: [params] <br>
     * @Return: long <br>
     * @Description：获取源表数据总量<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/25 19:25 <br>
     * @Version： V2.0.2<br>
     */
    private long getSourceTableCount(TableContrastParamEntity params) {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getImpalaSourceConnection();
            dataManagement.getStatusMap().get(params.getTableFlag()).setTableContrastStartTime(DateUtil.now());
            long sourceDataNum = impalaQueryService.queryTotalCount(connection, params.getSourceDbAndTableName()
                    , params.getSourceQueryCriteria(), params.getTableFlag());
            dataManagement.getStatusMap().get(params.getTableFlag()).setTableDataTotal(sourceDataNum);
            return sourceDataNum;
        } catch (Exception e) {
            throw new RuntimeException(StrUtil.format("源表数据总量获取异常: {}", e.getMessage()), e);
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    /*
     * @Method: getTargetTableCount <br>
     * @Param: [params] <br>
     * @Return: long <br>
     * @Description：获取目标表数据总量<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/25 19:25 <br>
     * @Version： V2.0.2<br>
     */
    private long getTargetTableCount(TableContrastParamEntity params) {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getImpalaTargetConnection();
            dataManagement.getStatusMap().get(params.getTableFlag()).setTableContrastStartTime(DateUtil.now());
            long targetDataNum = impalaQueryService.queryTotalCount(connection, params.getTargetDbAndTableName()
                    , params.getTargetQueryCriteria(), params.getTableFlag());
            dataManagement.getStatusMap().get(params.getTableFlag()).setTargetTableDataTotal(targetDataNum);
            return targetDataNum;
        } catch (Exception e) {
            throw new RuntimeException(StrUtil.format("对比表数据总量获取异常: {}", e.getMessage()), e);
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    /*
     * @Method: dataTotalCompare <br>
     * @Param: [params, sourceDataNum, targetDataNum] <br>
     * @Return: void <br>
     * @Description：源表和目标表数据总量对比<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/25 19:25 <br>
     * @Version： V2.0.2<br>
     */
    private void dataTotalCompare(TableContrastParamEntity params, long sourceDataNum, long targetDataNum) {
        if (sourceDataNum != targetDataNum) {
            throw new BizException(StrUtil.format("源表和对比表数据量不一致，源表数据量: {}, 对比表数据量: {}", sourceDataNum, targetDataNum));
        }
        dataManagement.getStatusMap().get(params.getTableFlag()).setTableContrastEndTime(DateUtil.now());
    }

    /*
     * @Method: checkStatisticsData <br>
     * @Param: [tableContrastParam] <br>
     * @Return: void <br>
     * @Description：统计数据对比<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/25 19:29 <br>
     * @Version： V2.0.2<br>
     */
    private void checkStatisticsData(TableContrastParamEntity params) {
        log.info("[数据比对组件]- 表[{}]表数据哈希值对比开始,开始时间: {}", params.getTableFlag(), DateUtil.now());
        Map<String, String> sourceTableStatisticsData = getSourceTableStatisticsData(params);
        // ② 获取目标表统计数据
        Map<String, String> targetTableStatisticsData = getTargetTableStatisticsData(params);
        // ③ 统计数据比对
        statisticsDataConstart(sourceTableStatisticsData, targetTableStatisticsData);
        log.info("[数据比对组件]- 表[{}]表数据哈希值对比结束,结束时间: {}", params.getTableFlag(), DateUtil.now());
    }

    /*
     * @Method: getSourceTableStatisticsData <br>
     * @Param: [params] <br>
     * @Return: java.util.Map<java.lang.String,java.lang.String> <br>
     * @Description：获取源表统计数据<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/26 17:44 <br>
     * @Version： V2.0.2<br>
     */
    private Map<String, String> getSourceTableStatisticsData(TableContrastParamEntity params) {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getImpalaTargetConnection();
            return impalaQueryService.queryTableStatisticsData(connection, params.getSourceDbAndTableName(), params.getSourceQueryCriteria(), params.getTableFlag());
        } catch (Exception e) {
            log.error("[数据比对组件] 源表[{}]统计哈希数据获取异常: {}", params.getTableFlag(), e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("源表统计哈希数据获取异常: {}", params.getTableFlag(), e.getMessage()));
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    /*
     * @Method: getTargetTableStatisticsData <br>
     * @Param: [params] <br>
     * @Return: java.util.Map<java.lang.String,java.lang.String> <br>
     * @Description：获取对比表统计数据<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/26 17:44 <br>
     * @Version： V2.0.2<br>
     */
    private Map<String, String> getTargetTableStatisticsData(TableContrastParamEntity params) {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getImpalaTargetConnection();
            return impalaQueryService.queryTableStatisticsData(connection, params.getTargetDbAndTableName(), params.getTargetQueryCriteria(), params.getTableFlag());
        } catch (Exception e) {
            log.error("[数据比对组件] 对比表[{}]统计哈希数据获取异常: {}", params.getTableFlag(), e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("对比表统计哈希数据获取异常: {}", params.getTableFlag(), e.getMessage()));
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    /*
     * @Method: statisticsDataConstart <br>
     * @Param: [sourceTableStatisticsData, targetTableStatisticsData] <br>
     * @Return: void <br>
     * @Description：数据比对-统计数据数值比对<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/26 17:44 <br>
     * @Version： V2.0.2<br>
     */
    private void statisticsDataConstart(Map<String, String> sourceTableStatisticsData, Map<String, String> targetTableStatisticsData) {
        if (sourceTableStatisticsData.size() != targetTableStatisticsData.size()) {
            throw new RuntimeException(StrUtil.format("统计哈希数据不一致, 源表统计数据: {}, 目标表统计数据: {}"
                    , sourceTableStatisticsData.toString(), targetTableStatisticsData.toString()));
        }
        for (Map.Entry<String, String> entry : sourceTableStatisticsData.entrySet()) {
            if (!entry.getValue().equals(targetTableStatisticsData.get(entry.getKey()))) {
                throw new RuntimeException(StrUtil.format("统计哈希数据不一致, 源表统计数据: {}, 目标表统计数据: {}"
                        , sourceTableStatisticsData.toString(), targetTableStatisticsData.toString()));
            }
        }
    }

    /*
     * @Method: checkDetailedData <br>
     * @Param: [tableContrastParam] <br>
     * @Return: void <br>
     * @Description：明细数据对比<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/26 19:18 <br>
     * @Version： V2.0.2<br>
     */
    private void checkDetailedData(TableContrastParamEntity tableContrastParam, long dataTotal) {
        log.info("[数据比对组件]- 表[{}]表明细数据对比开始,开始时间: {}", tableContrastParam.getTableFlag(), DateUtil.now());
        if ((dataContrastConstants.isItemizedDetailedDataConstart(dataTotal))) {
            itemizedDetailedDataConstart(tableContrastParam);
        } else {
            sampleDetailedDataConstart(tableContrastParam, dataTotal);
        }
        log.info("[数据比对组件]- 表[{}]表明细数据对比结束,结束时间: {}", tableContrastParam.getTableFlag(), DateUtil.now());
    }

    /*
     * @Method: sampleDetailedDataConstart <br>
     * @Param: [tableContrastParam, dataTotal] <br>
     * @Return: void <br>
     * @Description：抽样数据对比<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/27 16:52 <br>
     * @Version： V2.0.2<br>
     */
    private void sampleDetailedDataConstart(TableContrastParamEntity tableContrastParam, long dataTotal) {
        if (dataContrastConstants.isQuantifyDetailedDataConstart(dataTotal)) {
            quantifyDetailedDataConstart(tableContrastParam, dataTotal);
        } else if (dataContrastConstants.isRatioDetailedDataConstartDetailedDataConstart()) {
            ratioDetailedDataConstart(tableContrastParam, dataTotal);
        }
        dataManagement.setAchieveStatus(tableContrastParam.getTableFlag());
    }

    /*
     * @Method: ratioDetailedDataConstart <br>
     * @Param: [tableContrastParam] <br>
     * @Return: void <br>
     * @Description：比例数据比对<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/26 19:21 <br>
     * @Version： V2.0.2<br>
     */
    private void ratioDetailedDataConstart(TableContrastParamEntity params, long dataTotal) {
        try {
            long constartNum = dataTotal / dataContrastConstants.getSampleRatio() / dataContrastConstants.getSampleNumber();
            for (int i = 0; i < constartNum; i++) {
                appointDataConstart(params, dataTotal);
            }
        } catch (Exception e) {
            log.error("[数据比对组件] 表[{}]按比例明细数据对比异常: {}", params.getTableFlag(), e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("按比例明细数据对比异常: {}", e.getMessage()));
        }
    }

    /*
     * @Method: appointDataConstart <br>
     * @Param: [params, dataTotal] <br>
     * @Return: void <br>
     * @Description：随机指定起始位置和查询行数进行数据对比<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/27 16:16 <br>
     * @Version： V2.0.2<br>
     */
    private void appointDataConstart(TableContrastParamEntity params, long dataTotal) {
        String sqlTemplate = "SELECT * FROM {} {} LIMIT {} OFFSET {}";
        int offset = new Random().nextInt((int) (dataTotal - 1 + 1)) + 1;
        List sourceData = getSourceAppointDetailedData(params, sqlTemplate, offset);
        List targetData = getTargetAppointDetailedData(params, sqlTemplate, offset);
        if (!sourceData.equals(targetData)) {
            throw new RuntimeException(StrUtil.format("明细数据{}行 ☞ {}行中数据不一致", offset, (offset + dataContrastConstants.getSampleNumber())));
        }
    }

    /*
     * @Method: getSourceAppointDetailedData <br>
     * @Param: [params, sqlTemplate, offset] <br>
     * @Return: java.util.List <br>
     * @Description：随机指定起始位置和查询行数源表数据查询<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/27 16:16 <br>
     * @Version： V2.0.2<br>
     */
    private List getSourceAppointDetailedData(TableContrastParamEntity params, String sqlTemplate, int offset) {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getHiveSourceConnection();
            String querySql = StrUtil.format(sqlTemplate, params.getSourceDbAndTableName(),
                    params.getSourceQueryCriteria(), dataContrastConstants.getSampleNumber(), offset);
            return impalaQueryService.queryTableData(connection, querySql);
        } catch (Exception e) {
            log.error("[数据比对组件] 源表[{}]抽样数据比对查询数据异常: {}", params.getTableFlag(), e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("抽样数据比对查询数据异常: {}", e.getMessage()));
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    /*
     * @Method: getTargetAppointDetailedData <br>
     * @Param: [params, sqlTemplate, offset] <br>
     * @Return: java.util.List <br>
     * @Description：随机指定起始位置和查询行数对比表数据查询<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/27 16:16 <br>
     * @Version： V2.0.2<br>
     */
    private List getTargetAppointDetailedData(TableContrastParamEntity params, String sqlTemplate, int offset) {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getHiveTargetConnection();
            String querySql = StrUtil.format(sqlTemplate, params.getTargetDbAndTableName(),
                    params.getTargetQueryCriteria(), dataContrastConstants.getSampleNumber(), offset);
            return impalaQueryService.queryTableData(connection, querySql);
        } catch (Exception e) {
            log.error("[数据比对组件] 对比表[{}]抽样数据比对查询数据异常: {}", params.getTableFlag(), e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("抽样数据比对查询数据异常: {}", e.getMessage()));
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    /*
     * @Method: quantifyDetailedDataConstart <br>
     * @Param: [tableContrastParam] <br>
     * @Return: void <br>
     * @Description：定量数据比对<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/26 19:21 <br>
     * @Version： V2.0.2<br>
     */
    private void quantifyDetailedDataConstart(TableContrastParamEntity params, long dataTotal) {
        try {
            for (int i = 0; i < dataContrastConstants.getLargeAmountContrastTimes(); i++) {
                appointDataConstart(params, dataTotal);
            }
        } catch (Exception e) {
            log.error("[数据比对组件] 表[{}]按定量明细数据对比异常: {}", params.getTableFlag(), e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("表[{}]按定量明细数据对比异常: {}", params.getTableFlag(), e.getMessage()));
        }
    }

    /*
     * @Method: itemizedDetailedDataConstart <br>
     * @Param: [tableContrastParam] <br>
     * @Return: void <br>
     * @Description：逐条数据比对<br>
     * @Author： wz.li<br>
     * @Date： 2023/9/26 19:21 <br>
     * @Version： V2.0.2<br>
     */
    private void itemizedDetailedDataConstart(TableContrastParamEntity tableContrastParam) {
        dataManagement.getStatusMap().get(tableContrastParam.getTableFlag()).setDetailsContrastStartTime(DateUtil.now());
        //异步方法使用JDBC流式读取对比表中的明细数据
        dataReadHandler.readTargetDataToMap(tableContrastParam);
        //异步方法使用JDBC流式读取源表中的明细数据
        dataReadHandler.readSourceDataToQueue(tableContrastParam);
    }

    /**
     * @Method tableContrastFailHandler
     * @Param tableContrastParam
     * @Param errorMessage
     * @Return void
     * @Description 表比对失败处理
     * @Author Huang Yanyihe
     * @Date 2023/9/18 13:26
     * @Version V1.0
     */
    private void tableContrastFailHandler(TableContrastParamEntity tableContrastParam, String errorMessage) {
        dataManagement.setFailStatus(tableContrastParam.getTableFlag());
        dataManagement.setFalseMessage(tableContrastParam.getTableFlag(), errorMessage);
    }

    @Override
    public void cancelTableContrastTask(String tableFlag) {
        dataManagement.dataContrastError(tableFlag, -1L, "超时!");
    }
}
